#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <fcntl.h>
#include <signal.h> 
#include <stdlib.h>       
#include <string.h>
#include <pthread.h>
//#include <sys/sem.h>
#include <errno.h>
#include <sys/ipc.h>
//#include <sys/msg.h>
#include <sys/timeb.h>
#include <sys/socket.h>
//#include <sys/socketvar.h>
#include <sys/select.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <semaphore.h>
#include <string.h>
#include <termios.h>
#include <linux/input.h>
#include <semaphore.h>
#include <linux/shm.h>
#include <asm/unistd.h>
#include <sys/poll.h>

#include "common.h"


static const int sg_maxTotalRecvBufNum = 100 * 1000;


static void* ThreadRecv(void* arg);
static void* ThreadWork(void* arg);
//static void  restartProcess(char *process);
int startInternalTask(struct internalCaptureData *pIC);

struct internalCaptureData *createInternalTask(int port,const char *process,int bufCount,void(*pCallBack)(struct recvData *))
{
		struct internalCaptureData *icTemp;
		
		if( ( icTemp = (struct internalCaptureData *)malloc(sizeof(struct internalCaptureData)) ) == NULL )	
		{
				return NULL;	
		}		
		
		if( pthread_mutex_init (&(icTemp->Lock),NULL) != 0   ) 
		{
				printf("pthread_mutex_init failure {%s(%d)}\n",__FILE__,__LINE__);	
				return NULL;
		}
		
		if( sem_init(&(icTemp->taskSem), 1, 0) != 0   ) 
		{
				printf("sem_init failure {%s(%d)}\n",__FILE__,__LINE__);	
				return NULL;
		}
		
		if( ( icTemp->sock = UdpBind(port) ) < 0 )
		{
				printf("bind error!\n");	
				return NULL;
		}
		
		if( process )
		{
				strcpy(icTemp->process,process);
		}
		else
		{
				icTemp->process[0] = '\0';	
		}
		
		int i;
		struct recvData *back = NULL;
		struct recvData *temp = NULL;
		
		for( i = 0; i < bufCount; i++ )
		{
				if( ( temp = (struct recvData *)malloc(sizeof(struct recvData)) ) == NULL )	
				{
						return NULL;	
				}		
				
				temp->count = 0;
				temp->pNext = back;
				back = temp;
		}
		
		icTemp->emptyData = temp;	
		icTemp->recvData	= NULL;
		
		icTemp->pCallBack = pCallBack;
		
		icTemp->curTotalBufNum = bufCount;

		startInternalTask(icTemp);
		
		return icTemp;	
}

int startInternalTask(struct internalCaptureData *pIC)
{
		pthread_t ptID;
						
		if( pthread_create(&ptID, NULL, ThreadWork, (void *)pIC) != 0 )
		{
				printf("Make thread failure {%s(%d)}\n",__FILE__,__LINE__);		
				return -1;
		}	
		
		if( pthread_create(&ptID, NULL, ThreadRecv, (void *)pIC) != 0 )
		{
				printf("Make thread failure {%s(%d)}\n",__FILE__,__LINE__);		
				return -1;
		}						
		
		return 0;
}

static void* ThreadRecv(void* arg)
{
		int ret;
		socklen_t svrlen;
		struct pollfd pollflag;			
		struct recvData *recvDataTmp = NULL;	
		struct internalCaptureData *pIC = (struct internalCaptureData *)arg;
	
		while( !IsExit() )
		{
			  pollflag.fd      = pIC->sock;
			  pollflag.events  = POLLIN;
			  pollflag.revents = 0;	
			  
			  ret = poll( &pollflag, 1, 5000 );	
		
				if( ret == 0 ) 
				{
						//if(strlen(pIC->process)>0) restartProcess(pIC->process);
						continue;
				}	
				
				if( !(pollflag.revents) )	 
				{
						continue;
				}	
				
				if( recvDataTmp == NULL )
				{
						pthread_mutex_lock(&(pIC->Lock)); 
						recvDataTmp = GetHeadChain(&(pIC->emptyData));	
						pthread_mutex_unlock(&(pIC->Lock));
				}
				
				if( recvDataTmp == NULL ) //说明不够，从系统分配
				{
						if (pIC->curTotalBufNum >= sg_maxTotalRecvBufNum)
						{
							printf("this internalTask needs too much buf ! we do not give ! {%s(%d)}\n", __FILE__, __LINE__);
							return NULL;
						}

						printf("malloc memory!{%s(%d)}\n",__FILE__,__LINE__);	
						if( ( recvDataTmp = (struct recvData *)malloc(sizeof(struct recvData)) ) == NULL )	
						{
								printf("insufficient memory!{%s(%d)}\n",__FILE__,__LINE__);	
								return NULL;	
						}		
						pIC->curTotalBufNum++;
				}				
				
				recvDataTmp->count = 0;
				recvDataTmp->pNext = NULL;
				svrlen = sizeof(struct sockaddr_in);
				
				recvDataTmp->count = recvfrom(pIC->sock, recvDataTmp->buf, 1600, 0, (sockaddr*)&(recvDataTmp->addr), &svrlen);
				if( recvDataTmp->count < 1 )
				{
						continue;
				}

				pthread_mutex_lock(&(pIC->Lock)); 
				InsertTail(&(pIC->recvData),recvDataTmp);
				pthread_mutex_unlock(&(pIC->Lock));	
				
				sem_post(&(pIC->taskSem));
				
				//如果已经交给另外一个线程任务处理一定要调用，切记！
				recvDataTmp = NULL;

		}

		close(pIC->sock);	
	
		return NULL;	
}

static void* ThreadWork(void* arg)
{
		struct recvData *recvDataBak = NULL;
		struct recvData *recvDataTmp = NULL;
		struct internalCaptureData *pIC = (struct internalCaptureData *)arg;
		struct timespec ts;
		struct timeval  tv;						

		while( !IsExit() )
		{
				gettimeofday(&tv, NULL); 
				
			  ts.tv_sec  = tv.tv_sec + 5;
		    ts.tv_nsec = tv.tv_usec * 1000;		
		    
		    if( sem_timedwait(&(pIC->taskSem),&ts) != 0 )	continue;	
		    
		    pthread_mutex_lock(&(pIC->Lock)); 
				recvDataTmp = pIC->recvData;
				pIC->recvData = NULL;
				pthread_mutex_unlock(&(pIC->Lock));		

				while( recvDataTmp != NULL )
				{ 
						recvDataBak = recvDataTmp;
						recvDataTmp = recvDataTmp->pNext;
						recvDataBak->pNext = NULL;
						
						pIC->pCallBack(recvDataBak);
						
						recvDataBak->pNext = NULL;
						
						pthread_mutex_lock(&(pIC->Lock)); 
						InsertHead(&(pIC->emptyData),recvDataBak);	
						pthread_mutex_unlock(&(pIC->Lock));							
				}

		}		

		return NULL;	
}

/*
static void restartProcess(char *process)
{
		FILE *fp;
		int ifp=0,ret,count=0;
		char buf[1024];
		struct pollfd pollflag;	
		
		sprintf(buf,"ps | grep %s",process);

		if( ( fp = popen(buf,"r") ) == NULL )
		{
		   return;
		}
		
		ifp = fileno(fp);	
							
		fflush(fp);

		memset(buf, 0, 1024);
		if( ( ret = read(ifp,buf,1023) ) > 0 )	
		{						
				count = ret;
		}	

		pclose(fp);
						
		buf[count] = '\0';
		
		//printf("[%s](%d)\n\n",buf,count);

		if( strstr(buf,process) != NULL )
		{
				return;
		}	

		sprintf(buf,"%s incast\n",process);
		system(buf);
}
*/